package inventory

import (
	"bgs/db"
	sqlc "bgs/db/sqlc"
	"bgs/grpc/imp/constants"
	"bgs/grpc/imp/enums"
	nsqpUtil "bgs/nsqp/util"
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/nsqio/go-nsq"
	"go.uber.org/zap"
)

var (
	lastWarehouseUpsertMessage *WarehouseUpsertMessage = nil
)

// WarehouseUpsertMessage 接收仓库增加/修改的消息体
type WarehouseUpsertMessage struct {
	ID              int64   `json:"id"`
	SgID            int64   `json:"sg_id"`
	Name            string  `json:"name"`
	Address         string  `json:"address"`
	Latitude        float64 `json:"latitude"`
	Longitude       float64 `json:"longitude"`
	CityCode        int32   `json:"city_code"`
	RegionCode      string  `json:"region_code"`
	Contact         string  `json:"contact"`
	Mobile          string  `json:"mobile"`
	Pickup          bool    `json:"pick_up"`
	CityExpress     bool    `json:"city_express"`
	OrdinaryExpress bool    `json:"ordinary_express"`
	Deadline        string  `json:"deadline"`
	StorageMode     int32   `json:"storage_mode"`
	Timestamp       int64   `json:"timestamp"`
	Source          string  `json:"source"`
}

// WarehouseUpsertHandler inventory warehouse upsert message handler
type WarehouseUpsertHandler struct {
	dao      *db.PgDao
	slog     *zap.SugaredLogger
	topic    string
	failover *nsqpUtil.NsqFailover
}

// NewWarehouseUpsertHandler create a new handler for Warehouse upsert
func NewWarehouseUpsertHandler(
	pgdao *db.PgDao,
	slog *zap.SugaredLogger,
	topic string,
	failover *nsqpUtil.NsqFailover,
) *WarehouseUpsertHandler {
	return &WarehouseUpsertHandler{
		dao:      pgdao,
		slog:     slog.With("module", "WarehouseUpsertHandler"),
		topic:    topic,
		failover: failover,
	}
}

// HandleMessage implements the Handler interface.
func (h *WarehouseUpsertHandler) HandleMessage(m *nsq.Message) error {
	if len(m.Body) == 0 {
		// Returning nil will automatically send a FIN command to NSQ to mark the message as processed.
		return nil
	}

	h.slog.Infof("message=>%s", string(m.Body))

	v := &WarehouseUpsertMessage{}
	if decodeErr := json.NewDecoder(bytes.NewReader(m.Body)).Decode(v); decodeErr != nil {
		return fmt.Errorf("无法解析信息[WarehouseUpsertMessage]: %s", decodeErr.Error())
	}

	if lastWarehouseUpsertMessage != nil && lastWarehouseUpsertMessage.Timestamp == v.Timestamp {
		h.slog.Infof("过时的消息: lastWarehouseUpsertMessage.Timestamp:%v,v.Timestamp:%v", lastWarehouseUpsertMessage.Timestamp, v.Timestamp)
		return nil
	}

	if string(enums.AppIDBGAPI) == v.Source {
		h.slog.Infof("本地发起的消息不用处理:@%v v:%v", v.Timestamp, v)
		return nil
	}

	if v.Pickup && v.CityExpress && !v.OrdinaryExpress {
		h.slog.Infof("仓库仅支持同城配送,暂不同步:@%v v:%v", v.Timestamp, v)
		return nil
	}

	// 处理仓库新建或者修改的消息

	ctx := context.Background()
	err := h.dao.NewTxRunner().SqlcExecTx(ctx, func(q *sqlc.Queries) (err error) {

		brandID := v.SgID
		txTime := time.Now()
		localObj, localPkName, localFields := constants.SyncLocalObjBrandWarehouse, constants.SyncLocalPkNameBrandWarehouse, constants.SyncLocalFieldsBrandWarehouse
		remoteObj, remotePkName, remoteFields := constants.SyncRemoteObjBrandWarehouse, constants.SyncRemotePkNameBrandWarehouse, constants.SyncRemoteFieldsBrandWarehouse
		remotePk := v.ID

		var syncAction enums.EnumSyncAction
		var deliveryModes = []string{}
		if v.Pickup {
			deliveryModes = append(deliveryModes, string(enums.DeliveryModeSelfPickup))
		}
		if v.CityExpress {
			deliveryModes = append(deliveryModes, string(enums.DeliveryModeInCity))
		}
		if v.OrdinaryExpress {
			deliveryModes = append(deliveryModes, string(enums.DeliveryModeExpress))
		}
		var storeID int64 = 0
		var pStoreID int64 = 0
		var permissionID int64 = 0
		var sync sqlc.Sync
		var syncMapping sqlc.SyncMapping
		var localPk string

		brandWarehouse, sqlErr := q.GetBrandWarehouseByRemotePk(ctx, remotePk)
		isExist := true
		if sqlErr != nil {
			if sql.ErrNoRows != sqlErr {
				err = sqlErr
				return
			}
			isExist = false
		}

		if isExist {
			// 更新仓库信息
			syncAction = enums.SyncActionRemoteUpdateToLocal
			localPk = fmt.Sprintf("%d", brandWarehouse.ID)
			brandWarehouse, err = q.UpdateBrandWarehouse(ctx, sqlc.UpdateBrandWarehouseParams{
				ID:            brandWarehouse.ID,
				Name:          v.Name,
				Tel:           v.Mobile,
				Linkman:       v.Contact,
				Address:       v.Address,
				Lng:           strconv.FormatFloat(v.Longitude, 'E', -1, 64),
				Lat:           strconv.FormatFloat(v.Latitude, 'E', -1, 64),
				DeliveryModes: brandWarehouse.DeliveryModes,
				CityCode:      brandWarehouse.CityCode,
				Deadline:      v.Deadline,
				StorageMode:   v.StorageMode,
				StoreID:       storeID,
				PStoreID:      pStoreID,
				PermissionID:  permissionID,
			})
			if err != nil {
				return err
			}
			h.slog.Info("1 nsq更新品牌仓库 ok")

		} else {
			// 新增仓库信息
			syncAction = enums.SyncActionRemoteCreateToLocal
			brandWarehouse, err = q.CreateBrandWarehouse(ctx, sqlc.CreateBrandWarehouseParams{
				Origin:        string(enums.DataSourceRemote),
				BrandID:       brandID,
				Name:          v.Name,
				Tel:           v.Mobile,
				Linkman:       v.Contact,
				Address:       v.Address,
				Lng:           strconv.FormatFloat(v.Longitude, 'E', -1, 64),
				Lat:           strconv.FormatFloat(v.Latitude, 'E', -1, 64),
				DeliveryModes: deliveryModes,
				CityCode:      string(v.CityCode),
				Deadline:      v.Deadline,
				StorageMode:   v.StorageMode,
				StoreID:       storeID,
				PStoreID:      pStoreID,
				PermissionID:  permissionID,
				RemotePk:      remotePk,
			})
			if err != nil {
				return err
			}
			h.slog.Info("1 nsq创建品牌仓库 ok")

		}

		sync, err = q.UpsertSync(ctx, sqlc.UpsertSyncParams{
			LocalObj:         localObj,
			LocalPkName:      localPkName,
			LocalFields:      localFields,
			RemoteObj:        remoteObj,
			RemotePkName:     remotePkName,
			RemoteFields:     remoteFields,
			LatestSynced:     txTime,
			LatestSyncAction: string(syncAction),
		})
		if err != nil {
			return err
		}
		h.slog.Info("2 nsq仓库同步 ok")

		// 3 设置同步映射
		syncMapping, err = q.UpsertSyncMapping(ctx, sqlc.UpsertSyncMappingParams{
			SyncID:    sync.ID,
			LocalObj:  localObj,
			LocalPk:   localPk,
			RemoteObj: remoteObj,
			RemotePk:  strconv.FormatInt(remotePk, 10),
		})
		if err != nil {
			return err
		}
		h.slog.Info("3 nsq仓库同步映射 ok")

		syncData, err := json.Marshal(brandWarehouse)
		if err != nil {
			return err
		}

		maxVer, err := q.GetSyncLogMaxVer(ctx, sqlc.NInt64FromInt64(syncMapping.ID))
		if err != nil {
			return err
		}

		_, err = q.CreateSyncLog(ctx, sqlc.CreateSyncLogParams{
			SyncID:        sync.ID,
			SyncMappingID: sqlc.NInt64FromInt64(syncMapping.ID),
			LocalObj:      localObj,
			LocalPk:       localPk,
			RemoteObj:     remoteObj,
			RemotePk:      strconv.FormatInt(remotePk, 10),
			Synced:        txTime,
			Action:        string(syncAction),
			Data:          syncData,
			SyncStatus:    string(enums.SyncStatusDone),
			Ver:           int64(maxVer + 1),
		})
		if err != nil {
			return err
		}
		h.slog.Info("4 nsq仓库同步日志 ok")

		// 5 设置同步远程对象
		_, err = q.UpsertSyncRemote(ctx, sqlc.UpsertSyncRemoteParams{
			SyncID:    sync.ID,
			RemoteObj: remoteObj,
			RemotePk:  strconv.FormatInt(remotePk, 10),
			Data:      json.RawMessage(m.Body),
		})

		if err != nil {
			return err
		}
		h.slog.Info("5 nsq仓库同步远程对象 ok")

		return
	})

	if err != nil {
		return err
	}

	lastWarehouseUpsertMessage = v
	// Returning a non-nil error will automatically send a REQ command to NSQ to re-queue the message.
	return nil
}
